package oldcode;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;

import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.to8to.kitt.Constants;

public class ChannelPool {
	
	public static Logger logger = LoggerFactory.getLogger(ChannelPool.class);  
	
	private String service;
	
	private String instance;
	
	private int size = Constants.CHANNEL_WIDTH;
	
	private Bootstrap bootstrap;
	
	private String host;
	
	private int port;
	
	private Map<String, DownChannelState> channels = new ConcurrentHashMap<String, DownChannelState>();
	
	public Map<String, DownChannelState> getChannels() {
		
		return channels;
	}

	public String address()
	{
		return host + ":" + port;
	}

	public String toString()
	{
		StringBuilder sb = new StringBuilder();
		
		for(DownChannelState s : channels.values())
		{
			sb.append(s.toString() + "\n");
		}
		
		return sb.toString();
	}
	
	public ChannelPool(Bootstrap bootstrap, String host, int port, int size)
	{
		this.bootstrap = bootstrap;
		this.size = size;
		this.host = host;
		this.port = port;
		init();
	}
	
	
	public String getService() {
		return service;
	}

	public void setService(String service) {
		this.service = service;
	}

	public String getInstance() {
		return instance;
	}

	public void setInstance(String instance) {
		this.instance = instance;
	}

	public void close()
	{
		for(DownChannelState s : channels.values())
		{
			ChannelFuture cf = s.channel.close();
			
			cf.addListener(new ChannelFutureListener(){
				@Override
				public void operationComplete(ChannelFuture future)
						throws Exception {
					
					Channel channel = future.channel();
					
					if(future.isSuccess())
						logger.info("断开连接成功: {} {}", channel, service + "-" + instance);
					else 
						logger.error("断开连接失败: {} {}", channel, service + "-" + instance);
					
					future.removeListener(this);
				}
			});
		}
		
		channels.clear();
	}
	
	
	public boolean checkChannelHealth()
	{
		Set<String> brokenIds = new HashSet<String>();
		for(String id : channels.keySet())
		{
			DownChannelState s = channels.get(id);
			
			Channel channel = s.channel;
			
			boolean isOpen = channel.isOpen();
			
			boolean isActive = channel.isActive();
			
			boolean isRegistered = channel.isRegistered();
			
			boolean isWritable = channel.isWritable();
			
			if(!(isOpen && isActive && isRegistered && isWritable))
				brokenIds.add(id);
		}
		return brokenIds.size() == 0;
	}
	
	
	private void initOnce()
	{
		ChannelFuture f = bootstrap.connect(host, port);
		
		f.addListener(new ChannelFutureListener(){

			@Override
			public void operationComplete(ChannelFuture future)
					throws Exception {
				synchronized(channels)
				{
					
					Channel channel = future.channel();
	
					if(future.isSuccess())
					{
						logger.info("连接成功: {} {}", channel, service + "-" + instance);
						
						DownChannelState s = new DownChannelState();
						
						s.channel = channel;
						
						String id = id(channel);
						
						channels.put(id, s);
						
						future.removeListener(this);
					}
					else  
					{
						logger.error("连接失败, {} {}", channel, service + "-" + instance);
					}
			
				}
				future.removeListener(this);
			}
			
		});
	}
	
	public void reinit()
	{
		close();
		init();
	}
	
	public void init()
	{
		for(int i = 0 ; i < size ; ++ i)
			initOnce();
	}
	
	private static String id(Channel channel)
	{
		return channel.id().asLongText();
	}
	
	public void removePending(Channel channel, int seqid)
	{
		logger.trace("remove pending, channel:{}, seq:{}", channel, seqid);
		String id = id(channel);
		DownChannelState s = channels.get(id);
		if(s != null)
			s.pendingMsg.remove(seqid);
		else 
			logger.warn("#####channel state not found: " + channel.toString());
	}
	
	private void addPending(Channel channel, int seqid)
	{
		logger.trace("add pending, channel:{}, seq:{}", channel, seqid);
		
		String id = id(channel);
		
		DownChannelState s = channels.get(id);
		if(s != null)
			s.pendingMsg.put(seqid, 1);
		else 
			logger.warn("######channel state not found: " + channel.toString());
	}
	
	private Channel getChannel()
	{
		int size = channels.size();
		
		if(channels.size() == 0)
		{
			logger.warn("no channel connected!");
			return null;
		}
		
		List<DownChannelState> list = new LinkedList<DownChannelState>();
		
		list.addAll(channels.values());
		
		int index = ((int)System.nanoTime() % size + size)%size;
		
		DownChannelState s = list.get(index);
		
		return s.channel;
	}
	
	public void send(Object msg, int seqid, ChannelFutureListener l)
	{
		Channel channel = getChannel();
		logger.debug("<=== 开始发送 , 链路: {}", channel);
		if(channel != null)
		{
			addPending(channel, seqid);
            ChannelFuture cf = channel.writeAndFlush(msg);
//            ChannelFuture cf = channel.write(msg);
            if(l != null)
            	cf.addListener(l);
		}
		else
			logger.error("<=== 发送链路为空!!!! {}", channels);
	}
	
}
